package com.zjh.common.SSE.service.Impl;

import com.zjh.admin.entity.DAO.SysUserEntity;
import com.zjh.admin.service.UserService;
import com.zjh.common.SSE.entity.SSEDataBody;
import com.zjh.common.SSE.service.SSEService;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Service
public class SSEServiceImpl implements SSEService {

    public static final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    public static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    @Resource
    private UserService userService;

    @PostConstruct
    public void init() {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

        // 每1秒发送一次时间消息
        scheduler.scheduleAtFixedRate(this::sendTimeMessages, 0, 1, TimeUnit.SECONDS);
    }


    @Override
    public void registerSSEClient(String clientId, SseEmitter emitter) {
        // 注册连接
        emitters.put(clientId, emitter);

        SSEDataBody sseDataBody = paraSSEData(clientId);
        handleOnline(sseDataBody);

        // 连接关闭时的处理
        emitter.onCompletion(() -> {

            if (emitters.containsKey(clientId)) {
                emitters.remove(clientId);
                handleOffline(sseDataBody);
            }

        });
        emitter.onTimeout(() -> {
            if (emitters.containsKey(clientId)) {
                emitters.remove(clientId);
                handleOffline(sseDataBody);
            }

        });
        emitter.onError((e) -> {
            if (emitters.containsKey(clientId)) {
                emitters.remove(clientId);
                handleOffline(sseDataBody);
            }

        });
    }

    private void handleOffline(SSEDataBody sseDataBody) {
        // 下线提醒
        if ("SystemClock".equals(sseDataBody.getPageType())) {
            SysUserEntity loginUser = userService.getUserByUsername(sseDataBody.getUsername());
            sseDataBody.setData(loginUser);
            sseDataBody.setType("SomeOneOffline");

            sendMessagesToAll(sseDataBody);

            // 通知 在线用户页面 刷新用户列表
            sendRefreshOnlineUserToAllOnlinePage();
        }
    }

    private void handleOnline(SSEDataBody sseDataBody) {
        if ("SystemClock".equals(sseDataBody.getPageType())) {

            SysUserEntity loginUser = userService.getUserByUsername(sseDataBody.getUsername());
            sseDataBody.setData(loginUser);
            sseDataBody.setType("SomeOneOnline");

            sendMessagesToAll(sseDataBody);

            // 通知 在线用户页面 刷新用户列表
            sendRefreshOnlineUserToAllOnlinePage();
        }
    }

    private void sendRefreshOnlineUserToAllOnlinePage() {

        SSEDataBody body = new SSEDataBody();
        body.setPageType("SysOnlineUserPage");
        body.setType("refreshOnlineUserList");
        // 遍历所有客户端并发送消息
        emitters.forEach((clientId, emitter) -> {
            try {
                SSEDataBody origin = paraSSEData(clientId);
                // 在线用户界面
                if ("SysOnlineUserPage".equals(origin.getPageType())) {
                    emitter.send(SseEmitter.event().data(body));
                }

            } catch (IOException e) {
                // 发送失败，移除该客户端
                emitters.remove(clientId);
            }
        });

    }

    @Override
    public void sendMessagesToAll(SSEDataBody dataBody) {

        // 遍历所有客户端并发送消息
        emitters.forEach((clientId, emitter) -> {
            try {
                emitter.send(SseEmitter.event().data(dataBody));
            } catch (IOException e) {
                // 发送失败，移除该客户端
                emitters.remove(clientId);
            }
        });
    }

    @Override
    public void sendTimeMessages() {

        // 遍历所有客户端并发送消息
        emitters.forEach((clientId, emitter) -> {
            try {
                SSEDataBody<String> dataBody = new SSEDataBody<>();
                SSEDataBody origin = paraSSEData(clientId);

                String currentTime = LocalDateTime.now().format(formatter);
                dataBody.setData(currentTime);

                // 系统时钟连接
                if ("SystemClock".equals(origin.getPageType())) {

                    dataBody.setType("clockData");
                    emitter.send(SseEmitter.event().data(dataBody));
                }

            } catch (IOException e) {
                // 发送失败，移除该客户端
                emitters.remove(clientId);
            }
        });
    }

    @Override
    public String getClientId(String pageType, String type, String username, String clientIp) {
        return String.format("%s:%s:%s:%s", pageType, type, username, clientIp);
    }

    @Override
    public SSEDataBody paraSSEData(String clientId) {
        SSEDataBody sseDataBody = new SSEDataBody();
        String[] parts = clientId.split(":");

        sseDataBody.setPageType(parts[0]);
        sseDataBody.setType(parts[1]);
        sseDataBody.setUsername(parts[2]);
        sseDataBody.setIp(parts[3]);
        return sseDataBody;
    }
}
